package com.wyix;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 分布式任务调度器（简化版）
 * 使用并发容器实现线程安全的任务状态跟踪和结果存储
 */
public class DistributedTaskScheduler {
    // 任务状态枚举（待处理/处理中/完成）
    public enum TaskStatus {PENDING, RUNNING, COMPLETED}

    // 核心并发容器：线程安全的任务状态表（任务ID -> 状态）
    private final Map<String, TaskStatus> taskStatusMap = new ConcurrentHashMap<>();
    // 线程安全的任务结果表（任务ID -> 结果）
    private final Map<String, String> taskResultMap = new ConcurrentHashMap<>();
    // 模拟分布式节点ID生成器（原子操作保证线程安全）
    private final AtomicInteger nodeIdCounter = new AtomicInteger(1);

    // 任务执行线程池（模拟分布式执行节点集群）
    private final ExecutorService executor = new ThreadPoolExecutor(
            4,   // 核心线程数（常驻执行节点数）
            8,   // 最大线程数（突发流量时的最大节点数）
            30,  // 空闲线程存活时间（分钟）
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(100),  // 任务等待队列（最大排队任务数）
            r -> new Thread(r, "Scheduler-Node-" + nodeIdCounter.getAndIncrement()),  // 线程命名工厂
            new ThreadPoolExecutor.AbortPolicy()  // 队列满时拒绝新任务
    );

    /**
     * 提交任务到调度器
     *
     * @param task 具体任务逻辑（返回String类型的结果）
     * @return 生成的唯一任务ID
     */
    public String submitTask(Callable<String> task) {
        String taskId = UUID.randomUUID().toString().substring(0, 8);  // 生成短任务ID
        taskStatusMap.put(taskId, TaskStatus.PENDING);  // 初始状态设为待处理

        executor.submit(() -> {
            taskStatusMap.put(taskId, TaskStatus.RUNNING);  // 更新为处理中状态
            try {
                String result = task.call();  // 执行具体任务逻辑
                taskResultMap.put(taskId, result);  // 存储任务结果
            } catch (Exception e) {
                taskResultMap.put(taskId, "任务失败：" + e.getMessage());  // 异常处理
            } finally {
                taskStatusMap.put(taskId, TaskStatus.COMPLETED);  // 最终状态设为完成
            }
        });
        return taskId;
    }

    /**
     * 查询任务当前状态
     *
     * @param taskId 任务ID
     * @return 任务状态（默认返回PENDING）
     */
    public TaskStatus getTaskStatus(String taskId) {
        return taskStatusMap.getOrDefault(taskId, TaskStatus.PENDING);
    }

    /**
     * 获取任务执行结果
     *
     * @param taskId 任务ID
     * @return 任务结果（未完成时返回提示信息）
     */
    public String getTaskResult(String taskId) {
        return taskResultMap.getOrDefault(taskId, "任务未完成或不存在");
    }

    /**
     * 关闭调度器（释放资源）
     */
    public void shutdown() {
        executor.shutdown();
        try {
            if (!executor.awaitTermination(1, TimeUnit.MINUTES)) {
                executor.shutdownNow();
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
        }
    }

    // 示例用法
    public static void main(String[] args) throws InterruptedException {
        DistributedTaskScheduler scheduler = new DistributedTaskScheduler();

        // 模拟提交10个订单处理任务
        System.out.println("===== 提交任务 =====");
        for (int i = 0; i < 10; i++) {
            int orderId = i + 1;
            String taskId = scheduler.submitTask(() -> {
                Thread.sleep(1000);  // 模拟任务执行耗时（1秒）
                return "订单" + orderId + "处理完成（库存扣减成功）";
            });
            System.out.printf("提交订单%d，任务ID：%s\n", orderId, taskId);
        }

        // 轮询任务状态（模拟监控系统）
        System.out.println("\n===== 监控任务状态 =====");
        int completedCount = 0;
        while (completedCount < 10) {
            Thread.sleep(500);  // 每0.5秒检查一次
            for (Map.Entry<String, TaskStatus> entry : scheduler.taskStatusMap.entrySet()) {
                String taskId = entry.getKey();
                TaskStatus status = entry.getValue();
                if (status == TaskStatus.COMPLETED && !scheduler.getTaskResult(taskId).startsWith("任务失败")) {
                    completedCount++;
                    System.out.printf("任务%s完成，结果：%s\n", taskId, scheduler.getTaskResult(taskId));
                }
            }
        }

        // 关闭调度器
        scheduler.shutdown();
        System.out.println("\n===== 调度器已关闭 =====");
    }
}